package cn.datawin.cmppserver.netway.service;

import cn.datawin.cmppserver.queue.QueueCache;
import cn.datawin.sms.cmpp2.ReportLog;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;

@Slf4j
@Service
public class RocketMQService {


	@Autowired
	private DefaultMQProducer defaultMQProducer;

	/**
	 * 发送普通消息
	 */
	public void sendMessage(String topic, String tag, String message) {
		try {
			Message msg = new Message(topic, tag, message.getBytes(StandardCharsets.UTF_8));
			// 发送消息，同步方式
			var sendResult = defaultMQProducer.send(msg);
			log.info("消息发送成功，主题: {}, 标签: {}, 消息ID: {}", topic, tag, sendResult.getMsgId());
		} catch (Exception e) {
			log.error("消息发送失败，主题: {}, 标签: {}", topic, tag, e);
			throw new RuntimeException("消息发送失败", e);
		}
	}

	/**
	 * 发送异步消息
	 */
	public void sendAsyncMessage(String topic, String tag, String message) {
		try {
			Message msg = new Message(topic, tag, message.getBytes(StandardCharsets.UTF_8));
			defaultMQProducer.send(msg, new org.apache.rocketmq.client.producer.SendCallback() {
				@Override
				public void onSuccess(org.apache.rocketmq.client.producer.SendResult sendResult) {
					log.info("异步消息发送成功，主题: {}, 标签: {}, 消息ID: {}",
							topic, tag, sendResult.getMsgId());
				}

				@Override
				public void onException(Throwable e) {
					log.error("异步消息发送失败，主题: {}, 标签: {}", topic, tag, e);
				}
			});
		} catch (Exception e) {
			log.error("异步消息发送异常，主题: {}, 标签: {}", topic, tag, e);
			throw new RuntimeException("异步消息发送异常", e);
		}
	}

	/**
	 * 处理回执报告信息报告消息
	 * @param message
	 */
	public void dealReportMessage(String message) {
		ReportLog reportLog = JSON.parseObject(message, ReportLog.class);
		QueueCache.MESSAGES_REPORT_QUEUE.get(reportLog.getCstmCustomerCode()).offer(reportLog);
	}
}
